package com.nx.simple.producer;

import com.nx.Constant;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;


public class AsyncProducer {

        public static void main(String[] args) throws Exception {

            DefaultMQProducer producer = new DefaultMQProducer("nx-group");
            producer.setNamesrvAddr(Constant.MQ);
            producer.setSendMsgTimeout(10 * 1000);
            //设置发送失败重试机制(这里设置不重试 )
            producer.setRetryTimesWhenSendAsyncFailed(2);
            producer.start();
            int messageCount = 5;
            final CountDownLatch countDownLatch = new CountDownLatch(messageCount);
            for (int i = 0; i < messageCount; i++) {
                final int index = i;
                Message msg = new Message("nx-user",
                        "beijing",
                        "ms" + i,
                        "register info".getBytes(RemotingHelper.DEFAULT_CHARSET));
                //消息发送成功后，执行回调函数
                producer.send(msg, new SendCallback() {
                    @Override
                    public void onSuccess(SendResult sendResult) {
                        countDownLatch.countDown();
                        System.out.printf("%-10d OK %s %n", index,
                                sendResult.getMsgId());
                    }
                    @Override
                    public void onException(Throwable e) {
                        countDownLatch.countDown();
                        System.out.printf("%-10d Exception %s %n", index, e);
                        e.printStackTrace();
                    }
                });
            }
            countDownLatch.await(5, TimeUnit.SECONDS);

            producer.shutdown();
        }

}
